package cn.boat.router.utils;

import lombok.Data;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkBadVersionException;
import org.I0Itec.zkclient.exception.ZkMarshallingError;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.I0Itec.zkclient.serialize.ZkSerializer;
import org.apache.commons.lang.StringUtils;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Properties;

/**
 * Zookeeper工具类（Client）
 */
public class ZkClientUtils {
    public static Logger log = LoggerFactory.getLogger(ZkClientUtils.class);

    /**
     * 给定路径若不存在，则创建
     * @param client
     * @param path
     */
    public static void makeSurePersistentPathExists(final ZkClient client, final String path) {
        if (!client.exists(path)) {
            client.createPersistent(path, true);
        }
    }

    /**
     * 创建父路径
     * @param client
     * @param path
     */
    public static void createParentPath(final ZkClient client, final String path) {
        final String parentDir = path.substring(0, path.lastIndexOf('/'));
        if (parentDir.length() != 0) {
            client.createPersistent(parentDir, true);
        }
    }

    public static void createPersistentPath(final ZkClient client, final String path, final String data){
        try {
            client.createPersistent(path, data);
        }catch (Exception e) {
            updatePersistentPath(client, path, data);
        }
    }

    /**
     * 创建临时节点
     * @param client
     * @param path
     * @param data
     */
    public static void createEphemeralPath(ZkClient client, String path, String data) throws ZkNodeExistsException {
        try {
            client.createEphemeral(path, data);
        } catch (ZkNoNodeException e) {
            createParentPath(client, path);
            client.createEphemeral(path, data);
        }
    }



    /**
     * 创建临时节点，当节点存在则抛出异常
     * @param client
     * @param path
     * @param data
     */
    public static void createEphemeralPathExpectConflict(final ZkClient client, final String path, final String data) {
        try {
            createEphemeralPath(client, path, data);
        } catch (ZkNodeExistsException e) {

            // this canZkConfig happen when there is connection loss; make sure
            // the data
            // is what we intend to write
            String storedData = null;
            try {
                storedData = readData(client, path);
            } catch (final ZkNoNodeException e1) {
                // the node disappeared; treat as if node existed and let caller
                // handles this
            } catch (final Exception e2) {
                throw e2;
            }
            if (storedData == null || !storedData.equals(data)) {
                throw e;
            } else {
                // otherwise, the creation succeeded, return normally
                //logger.info(path + " exists with value " + data + " during connection loss");
            }
        } catch (final Exception e) {
            throw e;
        }

    }

    /**
     * 更新持久数据节点（CAS操作）
     * @param client
     * @param path
     * @param data
     */
    public static boolean updatePersistentPath(ZkClient client, String path, String data) {
        Stat stat = new Stat();
        while (true){
            try {
                client.readData(path, stat);
                // 为避免并发出错，修改数据时，要检验版本号
                client.writeData(path, data, stat.getVersion());
                return true;
            } catch (ZkBadVersionException e) {
                // 版本不对，进行重试
            } catch (ZkNoNodeException e) {
                // 节点不存在，则创建
                createParentPath(client, path);
                client.createPersistent(path, data);
                return true;
            } catch (Exception e) {
                e.printStackTrace();
                return false;
            }
        }

    }

    /**
     * 读取给定路径数据
     * @param client
     * @param path
     * @return
     */
    public static String readData(final ZkClient client, final String path) {
        return client.readData(path);
    }

    /**
     * 读取给定路径数据，若路径不存在返回null
     * @param client
     * @param path
     * @return
     */
    public static String readDataMaybeNull(final ZkClient client, final String path) {
        return client.readData(path, true);
    }

    /**
     * 更新临时节点数据，若不存在，则创建临时节点
     * @param client
     * @param path
     * @param data
     */
    public static boolean updateEphemeralPath(final ZkClient client, final String path, final String data) {
        Stat stat = new Stat();
        while (true) {
            try {
                client.readData(path, stat);
                // 为避免并发出错，修改数据时，要检验版本号
                client.writeData(path, data, stat.getVersion());
                return true;
            } catch (ZkBadVersionException e) {
                // 版本不对，进行重试
            } catch (final ZkNoNodeException e) {
                createParentPath(client, path);
                client.createEphemeral(path, data);
                return true;
            } catch (final Exception e) {
                throw e;
            }
        }

    }

    /**
     * 删除给定路径节点
     * @param client
     * @param path
     * @throws Exception
     */
    public static void deletePath(final ZkClient client, final String path) throws Exception {
        try {
            client.delete(path);
        } catch (final ZkNoNodeException e) {
            //logger.info(path + " : Node is exist.");
        } catch (final Exception e) {
            throw e;
        }
    }

    /**
     * 删除某个路径下所有节点
     * @param client
     * @param path
     * @throws Exception
     */
    public static void deletePathRecursive(final ZkClient client, final String path) throws Exception {
        try {
            client.deleteRecursive(path);
        } catch (final ZkNoNodeException e) {
            //logger.info(path + " : Node is exist.");

        } catch (final Exception e) {
            throw e;
        }
    }

    /**
     * 返回给定节点下所有子节点
     * @param client
     * @param path
     * @return
     */
    public static List<String> getChildren(final ZkClient client, final String path) {
        return client.getChildren(path);
    }

    /**
     * 返回给定节点下所有子节点，若节点不存在返回NUll
     * @param client
     * @param path
     * @return
     */
    public static List<String> getChildrenMaybeNull(final ZkClient client, final String path) {
        try {
            return client.getChildren(path);
        } catch (ZkNoNodeException e) {
            return null;
        }
    }

    /**
     * 判断节点是否存在
     * @param client
     * @param path
     * @return
     */
    public static boolean pathExists(final ZkClient client, final String path) {
        return client.exists(path);
    }

    /**
     * 获取父节点路径
     * @param path
     * @return
     */
    public static String getLastPart(final String path) {
        if (path == null) {
            return null;
        }
        final int index = path.lastIndexOf('/');
        if (index >= 0) {
            return path.substring(index + 1);
        } else {
            return null;
        }
    }



    @Data
    public static class ZKConfig extends Config implements Serializable {
        static final long serialVersionUID = -1L;

        /**
         * 根
         */
        @Key(name = "zk.zkRoot")
        public String zkRoot = "/boatMQ";
        /**
         * 是否启动zookeeper
         */
        @Key(name = "zk.zkEnable")
        public boolean zkEnable = true;

        /**
         * zookeeper主机
         */
        @Key(name = "zk.zkConnect")
        public String zkConnect = "127.0.0.1:2181";

        /**
         * zookeeper会话超时时间（默认30秒）
         */
        @Key(name = "zk.zkSessionTimeoutMs")
        public int zkSessionTimeoutMs = 30000;

        /**
         * 连接zookeeper最大容忍时间（默认30秒）
         */
        @Key(name = "zk.zkConnectionTimeoutMs")
        public int zkConnectionTimeoutMs = 30000;

        /**
         * zookeeper主从之间传输间隔时间
         */
        @Key(name = "zk.zkSyncTimeMs")
        public int zkSyncTimeMs = 5000;

        public ZKConfig() {
            loadZkConfigFromLocalFile();
        }

        public ZKConfig(final String zkConnect, final int zkSessionTimeoutMs, final int zkConnectionTimeoutMs,
                        final int zkSyncTimeMs) {
            super();
            loadZkConfigFromLocalFile();
            this.zkConnect = zkConnect;
            this.zkSessionTimeoutMs = zkSessionTimeoutMs;
            this.zkConnectionTimeoutMs = zkConnectionTimeoutMs;
            this.zkSyncTimeMs = zkSyncTimeMs;
        }

        public ZKConfig(final String zkRoot, final String zkConnect, final int zkSessionTimeoutMs,
                        final int zkConnectionTimeoutMs, final int zkSyncTimeMs, final boolean zkEnable) {
            super();
            loadZkConfigFromLocalFile();
            this.zkRoot = zkRoot;
            this.zkConnect = zkConnect;
            this.zkSessionTimeoutMs = zkSessionTimeoutMs;
            this.zkConnectionTimeoutMs = zkConnectionTimeoutMs;
            this.zkSyncTimeMs = zkSyncTimeMs;
            this.zkEnable = zkEnable;
        }

        public static long getSerialVersionUID() {
            return serialVersionUID;
        }

        public String getZkRoot() {
            return zkRoot;
        }

        public void setZkRoot(String zkRoot) {
            this.zkRoot = zkRoot;
        }

        public boolean isZkEnable() {
            return zkEnable;
        }

        public void setZkEnable(boolean zkEnable) {
            this.zkEnable = zkEnable;
        }

        public String getZkConnect() {
            return zkConnect;
        }

        public void setZkConnect(String zkConnect) {
            this.zkConnect = zkConnect;
        }

        public int getZkSessionTimeoutMs() {
            return zkSessionTimeoutMs;
        }

        public void setZkSessionTimeoutMs(int zkSessionTimeoutMs) {
            this.zkSessionTimeoutMs = zkSessionTimeoutMs;
        }

        public int getZkConnectionTimeoutMs() {
            return zkConnectionTimeoutMs;
        }

        public void setZkConnectionTimeoutMs(int zkConnectionTimeoutMs) {
            this.zkConnectionTimeoutMs = zkConnectionTimeoutMs;
        }

        public int getZkSyncTimeMs() {
            return zkSyncTimeMs;
        }

        public void setZkSyncTimeMs(int zkSyncTimeMs) {
            this.zkSyncTimeMs = zkSyncTimeMs;
        }

        void loadZkConfigFromLocalFile() {
            try {
                Properties properties = PropUtils.getResourceAsProperties("zookeeper.properties", "GBK");
                if (StringUtils.isNotBlank(properties.getProperty("zk.zkConnect"))) {
                    zkConnect = properties.getProperty("zk.zkConnect");
                }

                if (StringUtils.isNotBlank(properties.getProperty("zk.zkSessionTimeoutMs"))) {
                    zkSessionTimeoutMs = Integer.parseInt(properties.getProperty("zk.zkSessionTimeoutMs"));
                }

                if (StringUtils.isNotBlank(properties.getProperty("zk.zkConnectionTimeoutMs"))) {
                    zkConnectionTimeoutMs = Integer.parseInt(properties.getProperty("zk.zkConnectionTimeoutMs"));
                }

                if (StringUtils.isNotBlank(properties.getProperty("zk.zkSyncTimeMs"))) {
                    zkSyncTimeMs = Integer.parseInt(properties.getProperty("zk.zkSyncTimeMs"));
                }
            } catch (final IOException e) {
                log.error("zookeeper配置失败", e);
            }
        }
    }

    //字符串序列化
    public static class StringSerializer implements ZkSerializer {


        public Object deserialize(final byte[] bytes) throws ZkMarshallingError {
            try {
                return new String(bytes, "utf-8");
            } catch (final UnsupportedEncodingException e) {
                throw new ZkMarshallingError(e);
            }
        }


        public byte[] serialize(final Object data) throws ZkMarshallingError {
            try {
                return ((String) data).getBytes("utf-8");
            } catch (final UnsupportedEncodingException e) {
                throw new ZkMarshallingError(e);
            }
        }

    }

//    public static void main(String[] args) {
//        ZKConfig zkConfig = new ZKConfig();
//        Set<String> set = zkConfig.getFieldSet();
//        System.out.println(set);
//        System.out.println(zkConfig.findBestMatchField(set, "zkRoot"));
//    }
}
